package mq.cinrad.kafka.app.config;

import org.apache.commons.io.filefilter.AndFileFilter;
import org.apache.commons.io.filefilter.NotFileFilter;
import org.apache.commons.io.filefilter.OrFileFilter;
import org.apache.commons.io.filefilter.PrefixFileFilter;
import org.apache.commons.io.filefilter.SuffixFileFilter;
import org.apache.commons.io.filefilter.WildcardFileFilter;

import com.typesafe.config.Config;

public class ProducerConfigUtil {

	// private static final Logger logger =
	// LoggerFactory.getLogger(ConfigUtil.class);

	public static KafkaWatchFileConfig createKafkaWatchFileConfig(Config conf) {
		KafkaWatchFileConfig watchConfig = null;
		if (conf != null) {
			watchConfig = new KafkaWatchFileConfig();
			if (conf.hasPath("bootstrapServer"))
				watchConfig.setBootstrapServer(conf.getString("bootstrapServer"));
			if (conf.hasPath("defaultTopic"))
				watchConfig.setDefaultTopic(conf.getString("defaultTopic"));
			if (conf.hasPath("groupID"))
				watchConfig.setGroupID(conf.getString("groupID"));

			if (conf.hasPath("fileActionCallbackName"))
				watchConfig.setFileActionCallbackName(conf.getString("fileActionCallbackName"));

			if (conf.hasPath("notWildcardFileFilter"))
				watchConfig.setNotWildcardFileFilter(conf.getStringList("notWildcardFileFilter"));

			if (conf.hasPath("wildcardFileFilter"))
				watchConfig.setWildcardFileFilter(conf.getStringList("wildcardFileFilter"));

			if (conf.hasPath("prefixFileFilter"))
				watchConfig.setPrefixFileFilter(conf.getStringList("prefixFileFilter"));

			if (conf.hasPath("notPrefixFileFilter"))
				watchConfig.setNotPrefixFileFilter(conf.getStringList("notPrefixFileFilter"));

			if (conf.hasPath("suffixFileFilter"))
				watchConfig.setSuffixFileFilter(conf.getStringList("suffixFileFilter"));

			if (conf.hasPath("notSuffixFileFilter"))
				watchConfig.setNotSuffixFileFilter(conf.getStringList("notSuffixFileFilter"));

			if (conf.hasPath("nameRegex"))
				watchConfig.setNameRegex(conf.getString("nameRegex"));

			if (conf.hasPath("nameReplacement"))
				watchConfig.setNameReplacement(conf.getString("nameReplacement"));

			if (conf.hasPath("topicList"))
				watchConfig.setTopicList(conf.getStringList("topicList"));

			if (conf.hasPath("dirList"))
				watchConfig.setDirList(conf.getStringList("dirList"));
		}

		return watchConfig;
	}

	public static AndFileFilter buildAndFileFilter(KafkaWatchFileConfig kafkaWatchFileConfig) {
		AndFileFilter andFileFilter = new AndFileFilter();
		if (kafkaWatchFileConfig.getPrefixFileFilter() != null
				&& kafkaWatchFileConfig.getPrefixFileFilter().size() > 0) {
			PrefixFileFilter pf = new PrefixFileFilter(kafkaWatchFileConfig.getPrefixFileFilter());
			andFileFilter.addFileFilter(pf);
		}

		if (kafkaWatchFileConfig.getNotPrefixFileFilter() != null
				&& kafkaWatchFileConfig.getNotPrefixFileFilter().size() > 0) {
			PrefixFileFilter npf = new PrefixFileFilter(kafkaWatchFileConfig.getNotPrefixFileFilter());
			andFileFilter.addFileFilter(new NotFileFilter(npf));
		}

		if (kafkaWatchFileConfig.getSuffixFileFilter() != null
				&& kafkaWatchFileConfig.getSuffixFileFilter().size() > 0) {
			SuffixFileFilter sf = new SuffixFileFilter(kafkaWatchFileConfig.getSuffixFileFilter());
			andFileFilter.addFileFilter(sf);
		}

		if (kafkaWatchFileConfig.getNotSuffixFileFilter() != null
				&& kafkaWatchFileConfig.getNotSuffixFileFilter().size() > 0) {
			SuffixFileFilter nsf = new SuffixFileFilter(kafkaWatchFileConfig.getNotSuffixFileFilter());
			andFileFilter.addFileFilter(new NotFileFilter(nsf));
		}

		if (kafkaWatchFileConfig.getWildcardFileFilter() != null
				&& kafkaWatchFileConfig.getWildcardFileFilter().size() > 0) {
			WildcardFileFilter wf = new WildcardFileFilter(kafkaWatchFileConfig.getWildcardFileFilter());
			andFileFilter.addFileFilter(wf);
		}

		if (kafkaWatchFileConfig.getNotWildcardFileFilter() != null
				&& kafkaWatchFileConfig.getNotWildcardFileFilter().size() > 0) {
			WildcardFileFilter nwf = new WildcardFileFilter(kafkaWatchFileConfig.getNotWildcardFileFilter());
			andFileFilter.addFileFilter(new NotFileFilter(nwf));
		}

		return andFileFilter;
	}

	public static OrFileFilter buildOrFileFilter(KafkaWatchFileConfig kafkaWatchFileConfig) {
		OrFileFilter orFileFilter = new OrFileFilter();

		if (kafkaWatchFileConfig.getPrefixFileFilter() != null
				&& kafkaWatchFileConfig.getPrefixFileFilter().size() > 0) {
			PrefixFileFilter pf = new PrefixFileFilter(kafkaWatchFileConfig.getPrefixFileFilter());
			orFileFilter.addFileFilter(pf);
		}

		if (kafkaWatchFileConfig.getNotPrefixFileFilter() != null
				&& kafkaWatchFileConfig.getNotPrefixFileFilter().size() > 0) {
			PrefixFileFilter npf = new PrefixFileFilter(kafkaWatchFileConfig.getNotPrefixFileFilter());
			orFileFilter.addFileFilter(new NotFileFilter(npf));
		}

		if (kafkaWatchFileConfig.getSuffixFileFilter() != null
				&& kafkaWatchFileConfig.getSuffixFileFilter().size() > 0) {
			SuffixFileFilter sf = new SuffixFileFilter(kafkaWatchFileConfig.getSuffixFileFilter());
			orFileFilter.addFileFilter(sf);
		}

		if (kafkaWatchFileConfig.getNotSuffixFileFilter() != null
				&& kafkaWatchFileConfig.getNotSuffixFileFilter().size() > 0) {
			SuffixFileFilter nsf = new SuffixFileFilter(kafkaWatchFileConfig.getNotSuffixFileFilter());
			orFileFilter.addFileFilter(new NotFileFilter(nsf));
		}

		if (kafkaWatchFileConfig.getWildcardFileFilter() != null
				&& kafkaWatchFileConfig.getWildcardFileFilter().size() > 0) {
			WildcardFileFilter wf = new WildcardFileFilter(kafkaWatchFileConfig.getWildcardFileFilter());
			orFileFilter.addFileFilter(wf);
		}

		if (kafkaWatchFileConfig.getNotWildcardFileFilter() != null
				&& kafkaWatchFileConfig.getNotWildcardFileFilter().size() > 0) {
			WildcardFileFilter nwf = new WildcardFileFilter(kafkaWatchFileConfig.getNotWildcardFileFilter());
			orFileFilter.addFileFilter(new NotFileFilter(nwf));
		}

		return orFileFilter;
	}

}
